package com.namecard.namecardapi.service.impl;


import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.google.common.hash.BloomFilter;
import com.namecard.namecardapi.constants.BatchConstants;
import com.namecard.namecardapi.constants.ConstantsClass;
import com.namecard.namecardapi.entity.Industry;
import com.namecard.namecardapi.entity.LyUser;
import com.namecard.namecardapi.service.CountryService;
import com.namecard.namecardapi.service.IndustryService;
import com.namecard.namecardapi.service.LyUserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Component
@DS("sharding")
public class BatchProcessor {
    
    @Resource
    private LyUserService lyUserService;
    @Resource
    private CountryService countryService;
    @Resource
    private IndustryService industryService;

//    @Transactional
    public void processLyQueue(int batchSize, BloomFilter<String> bloomFilter) {
        BlockingQueue<LyUser> queue = BatchConstants.LY_QUEUE;
        batchSize = batchSize / 10;
        List<LyUser> buffer = new ArrayList<>(batchSize);

        try {
            while (true) {
                LyUser lyUser = queue.poll(200, TimeUnit.MILLISECONDS);
                if (lyUser != null) {
                    buffer.add(lyUser);
                }

                if (buffer.size() >= batchSize || (BatchConstants.LY_SIGNAL_COMPLETION.get() && !buffer.isEmpty() && queue.isEmpty())) {
                    //1、如果buffer队列满了，走这一步
                    //2、如果buffer队列没满，但是主线程已经读完数据了，并且阻塞队列已经空了，并且buffer还有剩余一点数据，走这一步
                    try {
                        //这里保存数据到领英表
                        List<Industry> industries = buffer.stream()
                                // 去重逻辑
                                .filter(ly -> {
                                    if (ly.getIndustry() == null) {
                                        return false;
                                    }
                                    if (bloomFilter.mightContain(ly.getIndustry())) {
                                        if (ly.getJobTitle() == null) {
                                            return false;
                                        } else if (bloomFilter.mightContain(ly.getJobTitle())) {
                                            return false;
                                        }
                                        return true;
                                    }
                                    bloomFilter.put(ly.getIndustry());
                                    return true;
                                })
                                .map(ly -> {
                            Industry industry = new Industry();
                            industry.setName(ly.getIndustry());
                            industry.setPositions(new ArrayList<>());
                            if (ly.getJobTitle() != null && !bloomFilter.mightContain(ly.getJobTitle())) {
                                bloomFilter.put(ly.getJobTitle());
                                industry.getPositions().add(ly.getJobTitle());
                            }
                            return industry;
                        }).collect(Collectors.toList());
                        industryService.batchInsertIndustry(industries);
                        lyUserService.saveBatch(buffer);
                        industries.clear();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    buffer.clear();
                    if (BatchConstants.LY_SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("批处理线程被中断", e);
        }
    }

}